package org.zfes.snowier.xyz.analysis.queue;

import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.zfes.snowier.common.eventpub.EventBusBucket;
import com.google.common.util.concurrent.ThreadFactoryBuilder;

public class MemoryAnalyQueueBucket implements AnalyQueueBucket{

	private Logger logger = LoggerFactory.getLogger(this.getClass());
	
//	private static SnowierQueue queue = new SnowierQueue(defaultTopic);
	
	private static  SnowierBlockQueue queue = new SnowierBlockQueue(defaultTopic);
	
	private static ExecutorService postExecutorService;
	/**
	 * 0.避免while-true检测，让队列有适当消息堆积能力
	 * 1.固定时间频率
	 * 2.队列超过长度
	 */
	private static ScheduledExecutorService scheduledPostExecutorService;
	
	
//	private static ExecutorService offerExecutorService;
	
	
	private volatile static AtomicBoolean runScheduled = new AtomicBoolean(false);
	 
	private volatile static AtomicBoolean sheduelPoolStarted = new AtomicBoolean(false);
	
	
	@Override
	public  void offer(Object msg) {
//		getOfferPool().submit(()->
		{
			queue.offer(msg);
			//logger.info("-after-offer-queue-size:{}",queue.size());
		}
//		);
		startPostSheduel();
	}
	
	@Override
	public  void clear() {
		queue.clear();
	}
	
	private synchronized void startPostSheduel(){
				if(sheduelPoolStarted.get()) {
					return;
				}
				
				if(postExecutorService==null) {
					postExecutorService=Executors.newSingleThreadExecutor();
				}
				if(scheduledPostExecutorService==null) {
					scheduledPostExecutorService=Executors.newSingleThreadScheduledExecutor(new ThreadFactoryBuilder().setNameFormat("-scheduled-post--pool-%d").build());
				}
				postExecutorService.submit(()->{
					while(true) {
						try {
							int size=queue.size();
							
							//logger.info("-postExecutor-runScheduled:{},queue-size:{}",runScheduled.get(),size);
							
							if(size > SCHEDULED_QUEUE_LENGTH) {
								runScheduled.set(false);
								doPoll(defaultTopic);
							}else {
								if(!runScheduled.get()) {
									runScheduled.set(true);
								}
							}
						}catch (Exception e) {
							e.printStackTrace();
						}
					}
				});
				
				scheduledPostExecutorService.scheduleAtFixedRate(()->{
					try {
						//logger.info("-scheduleAtFixedRate-runScheduled:{},queue-size:{}",runScheduled.get(),queue.size());
						if(runScheduled.get()) {
							doPoll(defaultTopic,batchSize);
						}
						
					}catch (Exception e) {
						e.printStackTrace();
					}
					
				},1,100,TimeUnit.MILLISECONDS);
				
			sheduelPoolStarted.set(true);
	
	}
	
	private void doPoll(String topic,int batchSize) {
		int beforeSize=queue.size();
		List<Object> dataList=queue.poll(batchSize);
		if(dataList!=null&&!dataList.isEmpty()) {
			//dataList.forEach(data->EventBusBucket.post(topic, new BasicMsg(data).setTopic(topic)));
			logger.info("-after-2-doPoll-before-size:{},-after-size:{}",beforeSize,queue.size());
		}
	}
	
	private void doPoll(String topic) {
		int beforeSize=queue.size();
		Object data=queue.poll();
		if(data!=null) {
		//	EventBusBucket.post(topic, new BasicMsg(data).setTopic(topic));
			logger.info("-after-1-doPoll-before-size:{},-after-size:{}",beforeSize,queue.size());
		}
	}
	
//	private static ExecutorService  getOfferPool() {
//		if(offerExecutorService != null) {
//			return offerExecutorService;
//		}
//		ThreadFactory namedThreadFactory = new ThreadFactoryBuilder().setNameFormat("-queueBucket-OfferPool-pool-%d").build();
//		offerExecutorService = new ThreadPoolExecutor(1, maxOfferThreads,0L, 
//						TimeUnit.MILLISECONDS,
//						new LinkedBlockingQueue<Runnable>(maxOfferThreadQueueSize), 
//						namedThreadFactory, 
//						new ThreadPoolExecutor.AbortPolicy());
//		return offerExecutorService;				
//		
//	}
	
}
